// Copyright (c) 2022 by Duguang.IO Inc. All Rights Reserved.
// Author: Ethan Liu
// Date: 2022-05-10 18:42:48

package client

import (
	"bytes"
	"context"
	"crypto/tls"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/google/uuid"
	"github.com/sirupsen/logrus"
	"io"
	"io/ioutil"
	"net/http"
	"path"
	"strings"
	"time"

	"jianmu-worker-kube/engine"
)

const (
	apiJoin    = "/workers/%s/join"
	apiPing    = "/workers/%s/ping"
	apiRequest = "/workers/%s/tasks"
	apiAccept  = "/workers/%s/tasks/%s/accept"
	apiFind    = "/workers/%s/tasks/%s"
	apiUpdate  = "/workers/%s/tasks/%s"
	apiBatch   = "/workers/%s/tasks/%s/logs/batch"
	apiUpload  = "/workers/%s/tasks/%s/logs"
)

// HTTPClient Worker API
type HTTPClient struct {
	Client     *http.Client
	Endpoint   string
	Secret     string
	WorkerId   string
	SkipVerify bool
}

type Task struct {
	Type           string `json:"type"`
	TaskInstanceId string `json:"taskInstanceId"`
	Pullstrategy   string `json:"pullStrategy"`
	Containerspec  struct {
		Image        string                   `json:"image"`
		Network      string                   `json:"network"`
		Networks     interface{}              `json:"networks"`
		Labels       interface{}              `json:"labels"`
		WorkingDir   string                   `json:"working_dir"`
		User         interface{}              `json:"user"`
		Host         interface{}              `json:"host"`
		Sock         interface{}              `json:"sock"`
		Environment  map[string]string        `json:"environment"`
		Secrets      []map[string]interface{} `json:"secrets"`
		Entrypoint   []string                 `json:"entrypoint"`
		Args         []string                 `json:"args"`
		VolumeMounts []struct {
			Source string `json:"source"`
			Target string `json:"target"`
		} `json:"volume_mounts"`
		Privileged   interface{} `json:"privileged"`
		ShmSize      interface{} `json:"shm_size"`
		DNS          interface{} `json:"dns"`
		DNSSearch    interface{} `json:"dns_search"`
		ExtraHosts   interface{} `json:"extra_hosts"`
		CPUPeriod    interface{} `json:"cpu_period"`
		CPUQuota     interface{} `json:"cpu_quota"`
		CPUShares    interface{} `json:"cpu_shares"`
		CPUSet       interface{} `json:"cpu_set"`
		MemswapLimit interface{} `json:"memswap_limit"`
		MemLimit     interface{} `json:"mem_limit"`
		Devices      interface{} `json:"devices"`
		Detach       interface{} `json:"detach"`
	} `json:"containerSpec"`
	Resultfile string `json:"resultFile"`
	Volume     struct {
		Name string `json:"name"`
		Type string `json:"type"`
	} `json:"volume"`
	Auth struct {
		Address  interface{} `json:"address"`
		Username interface{} `json:"username"`
		Password interface{} `json:"password"`
	} `json:"auth"`
	WorkData string `json:"workData"`
	Version  int    `json:"version"`
}

// New returns a new worker client.
func New(endpoint, secret string, skipverify bool, workerId string) *HTTPClient {
	client := &HTTPClient{
		Endpoint:   endpoint,
		Secret:     secret,
		WorkerId:   workerId,
		SkipVerify: skipverify,
	}
	if skipverify {
		client.Client = &http.Client{
			CheckRedirect: func(*http.Request, []*http.Request) error {
				return http.ErrUseLastResponse
			},
			Transport: &http.Transport{
				Proxy: http.ProxyFromEnvironment,
				TLSClientConfig: &tls.Config{
					InsecureSkipVerify: true,
				},
			},
		}
	}
	return client
}

var _ Client = (*HTTPClient)(nil)

// defaultClient is the default http.Client.
var defaultClient = &http.Client{
	CheckRedirect: func(*http.Request, []*http.Request) error {
		return http.ErrUseLastResponse
	},
}

// Ping sends a ping message to the server to test connectivity.
func (p *HTTPClient) Ping(ctx context.Context) error {
	uri := fmt.Sprintf(apiPing, p.WorkerId)
	_, err := p.do(ctx, uri, "GET", nil, nil, true)
	return err
}

func (p *HTTPClient) Join(ctx context.Context, worker *engine.Worker) error {
	uri := fmt.Sprintf(apiJoin, p.WorkerId)
	src := worker
	_, err := p.do(ctx, uri, "PUT", src, nil, true)
	return err
}

// Request 请求新任务执行 .
func (p *HTTPClient) Request(ctx context.Context, args *Filter) (*engine.Unit, error) {
	uri := fmt.Sprintf(apiRequest, p.WorkerId)
	src := args
	//dst := new(engine.Unit)
	dst := &Task{}
	_, err := p.retry(ctx, uri, "GET", src, dst, true)
	if err != nil {
		logrus.WithError(err).
			Error("Request error")
		return nil, err
	}

	return response2Unit(ctx, p.WorkerId, dst)
}

func response2Unit(ctx context.Context, workerId string, dst *Task) (*engine.Unit, error) {
	logrus.Debugf(" -----------> workerData: %s, Entrypoint: %s Args: %v  \n", dst.WorkData,
		dst.Containerspec.Entrypoint, dst.Containerspec.Args)

	var envs = make(map[string]string, 0)
	for _, secret := range dst.Containerspec.Secrets {
		decoder := base64.NewDecoder(base64.StdEncoding, strings.NewReader(secret["data"].(string)))
		content, err := ioutil.ReadAll(decoder)
		if err != nil {
			return nil, err
		}
		envs[secret["env"].(string)] = string(content)
	}

	for k, v := range dst.Containerspec.Environment {
		envs[k] = v
	}

	param := make([]string, 0)
	var ep string
	if len(dst.Containerspec.Entrypoint) > 0 {
		ep = dst.Containerspec.Entrypoint[0]
		if len(dst.Containerspec.Entrypoint) > 1 {
			param = append(param, dst.Containerspec.Entrypoint[1:]...)
		}
	}

	if len(dst.Containerspec.Args) > 0 {
		param = append(param, dst.Containerspec.Args...)
	}

	helperImage := ctx.Value("helperImage")
	helperImageName := helperImage.(string)

	runners := make([]*engine.Runner, 0)
	// VOLUME task has not image
	image := dst.Containerspec.Image
	if image == "" {
		image = helperImageName
	}
	build := engine.Runner{
		ID:           dst.TaskInstanceId,
		Version:      0,
		Command:      param,
		Detach:       false,
		DependsOn:    nil,
		Entrypoint:   ep,
		Envs:         envs,
		IgnoreStdout: false,
		IgnoreStderr: false,
		Image:        image,
		Name:         "build",
		Placeholder:  "",
		Privileged:   false,
		Resources:    nil,
		Pull:         0,
		Secrets:      nil,
		SpecSecrets:  nil,
		User:         nil,
		Group:        nil,
		Volumes:      nil,
		WorkingDir:   dst.Containerspec.WorkingDir,
		Cmd:          param,
	}
	runners = append(runners, &build)

	helper := engine.Runner{
		ID:           dst.TaskInstanceId,
		Version:      0,
		Command:      []string{"sh", "-c", "trap 'echo success;exit 0;' INT HUP QUIT TERM ; while :; do sleep 1; done"},
		Detach:       false,
		DependsOn:    nil,
		Entrypoint:   "",
		Envs:         dst.Containerspec.Environment,
		IgnoreStdout: false,
		IgnoreStderr: false,
		Image:        helperImageName,
		Name:         "helper",
		Placeholder:  "",
		Privileged:   false,
		Resources:    nil,
		Pull:         0,
		Secrets:      nil,
		SpecSecrets:  nil,
		User:         nil,
		Group:        nil,
		Volumes:      nil,
		WorkingDir:   "",
	}

	runners = append(runners, &helper)

	guid := uuid.New()
	uuid := guid.String()
	uuid = strings.ReplaceAll(uuid, "-", "")[:10]

	// TODO  nodeSelector
	nodeName := ctx.Value("nodeName")
	nodeNameStr, ok := nodeName.(string)
	var nodeSelector map[string]string
	if dst.WorkData != "" {
		logrus.WithField("workerData", dst.WorkData).
			WithField("instanceId", dst.TaskInstanceId).
			WithField("workerId", workerId).
			Debug("workerData describe")
		nodeSelector = map[string]string{
			"kubernetes.io/hostname": dst.WorkData,
		}
	} else {
		if ok {
			logrus.WithField("instanceId", dst.TaskInstanceId).
				WithField("workerId", workerId).
				Warn(fmt.Sprintf("workerData not found , use default node name: %s", nodeNameStr))
			if nodeNameStr == "" {
				logrus.WithField("instanceId", dst.TaskInstanceId).
					Infoln("default node is nil")
			} else {
				nodeSelector = map[string]string{
					"kubernetes.io/hostname": nodeNameStr,
				}
			}
		} else {
			logrus.Warn("config error JIANMU_WORKER_DEFAULT_NODE_NAME not config")
		}
	}

	k8sNode := false
	if dst.Resultfile == "" {
		if envs["JM_RESULT_FILE"] != "" {
			k8sNode = true
			dst.Resultfile = envs["JM_RESULT_FILE"]
		}
	}

	name := dst.Volume.Name
	if name == "" && len(dst.Containerspec.VolumeMounts) > 0 {
		name = dst.Containerspec.VolumeMounts[0].Source
	}
	name = path.Join("/", name)

	var secrets []*engine.Secret

	if dst.Containerspec.Secrets != nil {
		secrets = make([]*engine.Secret, 0)
		for _, s := range dst.Containerspec.Secrets {
			secrets = append(secrets, &engine.Secret{
				Name: s["env"].(string),
				Data: s["data"].(string),
				Mask: s["mask"].(bool),
			})
		}
	}

	return &engine.Unit{
		K8sNode: k8sNode,
		Type:    dst.Type,
		PodSpec: &engine.PodSpec{
			Name:               "jm-" + dst.TaskInstanceId + "-" + uuid,
			Namespace:          ctx.Value("nameSpace").(string),
			Annotations:        nil,
			Labels:             nil,
			NodeName:           "",
			NodeSelector:       nodeSelector,
			Tolerations:        nil,
			ServiceAccountName: "",
			HostAliases:        nil,
			DnsConfig:          engine.DnsConfig{},
		},
		Volume: &engine.Volumes{
			Name: name,
			Type: dst.Volume.Type,
		},
		Secrets:        secrets,
		PullSecret:     nil,
		Internal:       nil,
		Runners:        runners,
		TaskInstanceId: dst.TaskInstanceId,
		ResultFile:     dst.Resultfile,
		Resources:      nil,
		WorkerId:       workerId,
		WorkerData:     nodeSelector["kubernetes.io/hostname"],
	}, nil
}

// Accept 确认任务执行，乐观锁
func (p *HTTPClient) Accept(ctx context.Context, runner *engine.Runner) error {
	uri := fmt.Sprintf(apiAccept, p.WorkerId, runner.ID)
	src := runner
	dst := new(engine.Runner)
	_, err := p.retry(ctx, uri, "PATCH", src, dst, true)
	src.Version = dst.Version
	return err
}

// FindById 根据任务ID查找任务
func (p *HTTPClient) FindById(ctx context.Context, id string) (*engine.Unit, error) {
	uri := fmt.Sprintf(apiFind, p.WorkerId, id)
	dst := &Task{}
	_, err := p.retry(ctx, uri, "GET", nil, dst, true)
	if err != nil {
		logrus.WithError(err).
			WithField("id", id).
			Error("FindById Request error")
		return nil, err
	}
	return response2Unit(ctx, p.WorkerId, dst)
}

// Batch 实时上传日志.
func (p *HTTPClient) Batch(ctx context.Context, taskId string, lines []*engine.Line) error {
	uri := fmt.Sprintf(apiBatch, p.WorkerId, taskId)
	if taskId == "" {
		logrus.WithField("taskId", taskId).
			Warn("taskId is empty")
		return nil
	}
	_, err := p.do(ctx, uri, "POST", &lines, nil, false)
	return err
}

// Update 更新任务状态
func (p *HTTPClient) Update(ctx context.Context, task *engine.Task) error {
	uri := fmt.Sprintf(apiUpdate, p.WorkerId, task.TaskId)
	_, err := p.retry(ctx, uri, "PATCH", task, nil, true)
	return err
}

// Watch 监控任务是否取消
func (p *HTTPClient) Watch(ctx context.Context, taskId string) (bool, error) {
	uri := fmt.Sprintf(apiUpdate, p.WorkerId, taskId)
	res, err := p.retry(ctx, uri, "POST", nil, nil, true)
	if err != nil {
		return false, err
	}
	if res.StatusCode == 200 {
		return true, nil
	}
	return false, nil
}

// Upload 上传完整日志
func (p *HTTPClient) Upload(ctx context.Context, taskId string, lines []*engine.Line) error {
	uri := fmt.Sprintf(apiUpload, p.WorkerId, taskId)
	if lines == nil || len(lines) == 0 {
		return nil
	}
	_, err := p.retry(ctx, uri, "POST", &lines, nil, false)
	return err
}

func (p *HTTPClient) retry(ctx context.Context, path, method string, in, out interface{}, printTraceLog bool) (*http.Response, error) {
	for {
		res, err := p.do(ctx, path, method, in, out, printTraceLog)
		// do not retry on Canceled or DeadlineExceeded
		if err := ctx.Err(); err != nil {
			return res, err
		}
		// do not retry on optimisitic lock errors
		if err == ErrOptimisticLock {
			return res, err
		}
		if res != nil {
			// Check the response code. We retry on 500-range
			// responses to allow the server time to recover, as
			// 500's are typically not permanent errors and may
			// relate to outages on the server side.
			if res.StatusCode > 501 {
				time.Sleep(time.Second * 10)
				continue
			}
			// We also retry on 204 no content response codes,
			// used by the server when a long-polling request
			// is intentionally disconnected and should be
			// automatically reconnected.
			if res.StatusCode == 204 {
				time.Sleep(time.Second * 1)
				continue
			}
		} else if err != nil {
			time.Sleep(time.Second * 10)
			continue
		}

		return res, err
	}
}

// do is a helper function that posts a signed http request with
// the input encoded and response decoded from json.
func (p *HTTPClient) do(ctx context.Context, path, method string, in, out interface{}, printTraceLog bool) (*http.Response, error) {
	var buf bytes.Buffer

	// marshal the input payload into json format and copy
	// to an io.ReadCloser.
	if in != nil {
		json.NewEncoder(&buf).Encode(in)
	}

	endpoint := p.Endpoint + path
	req, err := http.NewRequest(method, endpoint, &buf)
	if err != nil {
		return nil, err
	}
	req = req.WithContext(ctx)

	req.Header.Set("Content-Type", "application/json; charset=utf-8")
	//req.Header.Set("Accept", "application/json, text/plain, */*")
	// the request should include the secret shared between
	// the agent and server for authorization.
	req.Header.Add("X-Jianmu-Token", p.Secret)

	if printTraceLog {
		logrus.Tracef("work http request method:%s,path:%s,body:%s\n", method, path, buf.String())
	}

	res, err := p.client().Do(req)
	if res != nil {
		defer func() {
			// drain the response body so we can reuse
			// this connection.
			io.Copy(ioutil.Discard, io.LimitReader(res.Body, 4096))
			res.Body.Close()
		}()
	}

	if err != nil {
		return res, err
	}

	if printTraceLog {
		logrus.WithField("statusCode", res.StatusCode).Traceln("http response status")
	}

	// if the response body return no content we exit
	// immediately. We do not read or unmarshal the response
	// and we do not return an error.
	if res.StatusCode == 204 {
		return res, nil
	}

	// Check the response for a 409 conflict. This indicates an
	// optimistic lock error, in which case multiple clients may
	// be attempting to update the same record. Convert this error
	// code to a proper error.
	if res.StatusCode == 409 {
		return nil, ErrOptimisticLock
	}

	// else read the response body into a byte slice.
	body, err := ioutil.ReadAll(res.Body)
	if err != nil {
		return res, err
	}

	if printTraceLog {
		logrus.WithField("response content", string(body)).Traceln("http response body")
	}

	if res.StatusCode > 299 {
		// if the response body includes an error message
		// we should return the error string.
		if len(body) != 0 {
			return res, errors.New(
				string(body),
			)
		}
		// if the response body is empty we should return
		// the default status code text.
		return res, errors.New(
			http.StatusText(res.StatusCode),
		)
	}
	if out == nil {
		return res, nil
	}

	return res, json.Unmarshal(body, out)
}

// client is a helper funciton that returns the default client
// if a custom client is not defined.
func (p *HTTPClient) client() *http.Client {
	if p.Client == nil {
		return defaultClient
	}
	return p.Client
}
